package com.dyna.utils;

import cn.hutool.extra.spring.SpringUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.ListenableFuture;

/**
 * kafka工具类
 * @author zhoucaiwang
 * @date 2024/5/14
 */
@Slf4j
public class KafkaUtil {

    private static KafkaTemplate<String, String> kafkaTemplate;

    static {
        KafkaTemplate bean = SpringUtil.getBean(KafkaTemplate.class);
        KafkaUtil.kafkaTemplate = bean;
    }

    /**
     * kafka发送消息
     * @param topic 主题
     * @param msg 消息
     */
    public static void send(String topic, String msg) {
        ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send(topic, msg);
        KafkaUtil.sendResult(future);
    }

    /**
     * kafka发送消息
     * @param topic 主题
     * @param msg 消息
     */
    public static void send(String topic, Integer  partition, String key, String msg) {
        ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send(topic, partition, key, msg);
        KafkaUtil.sendResult(future);
    }


    /**
     * kafka发送消息结果
     * @param future 发送结果
     */
    private static void sendResult(ListenableFuture<SendResult<String, String>> future) {
        future.addCallback(
                result -> {
                    if (result != null && result.getRecordMetadata()  != null) {
                        log.info("\n--->发送成功 - 主题: {}, - 分区: {}, 元数据: {}",
                                result.getRecordMetadata().topic(),
                                result.getRecordMetadata().partition(),
                                result.getProducerRecord().value());
                    } else {
                        log.error("发送成功但未获取到元数据信息");
                    }
                },
                ex -> log.error("发送失败: " + ex.getMessage())
        );
    }

}
